package com.hk.migration.cannal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.common.collect.Lists;
import com.hk.migration.cannal.process.TchListProcess;
import com.hk.migration.mapper.craftsman.InitMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@AllArgsConstructor
public class Consumer implements CommandLineRunner {

    private final InitMapper initMapper;

    private final TchListProcess tchListProcess;

    @Override
    public void run(String... args){
            init();
        /**
         * 加载处理处理队列
         */
        List<CanalProcess> canalProcessList = Arrays.asList(tchListProcess);

        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("116.62.19.236"
                ,10101), "example", null, null);
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            Runtime.getRuntime().addShutdownHook(new Thread(connector::disconnect));
            Executors.newScheduledThreadPool(1).execute(() -> {
                while (true){
                    try {
                        Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                        long batchId = message.getId();
                        int size = message.getEntries().size();
                        if(batchId != -1 || size > 0) {
                            List<CanalEntry.Entry> entry = message.getEntries();
//                                printSql(entry);
                            dispatcher(entry,canalProcessList);
                        }
                        connector.ack(batchId); // 提交确认
                    }catch (Exception ignored){
                    }finally {
                        try {
                            TimeUnit.SECONDS.sleep(1);
                        } catch (InterruptedException ignored) {
                        }
                    }
                }
        });
        } catch (Exception e) {
            log.error("e : {}",e);
            connector.disconnect();
        }
    }

    /**
     * 初始化
     */
    private void init() {
        List<String> list = initMapper.selectAllTable();
        list.forEach(s -> {
            s = "alter table "+s+" add column hk_id BIGINT not null  default 0 comment 'hello kid后台数据ID'";
            try {
                initMapper.addColumn(s);
            }catch (Exception ignored){
            }
        });
    }

    /**
     * 处理数据
     * @param entryList
     * @param canalProcessList
     */
    private void dispatcher(List<CanalEntry.Entry> entryList, List<CanalProcess> canalProcessList) {
        for (CanalEntry.Entry entry : entryList) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            CanalEntry.EventType eventType = rowChage.getEventType();
            String tableName = entry.getHeader().getTableName();
            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                List<CanalData> beforeCanalDataList = Lists.newLinkedList();
                List<CanalData> afterCanalDataList = Lists.newLinkedList();
                if (eventType == CanalEntry.EventType.DELETE) {
                    beforeCanalDataList = buildColumn(tableName,rowData.getBeforeColumnsList());
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    afterCanalDataList = buildColumn(tableName,rowData.getAfterColumnsList());
                } else {
                    beforeCanalDataList = buildColumn(tableName,rowData.getBeforeColumnsList());
                    afterCanalDataList = buildColumn(tableName,rowData.getAfterColumnsList());
                }
                try {
                    for (CanalProcess canalProcess : canalProcessList) {
                        if (canalProcess.apply(tableName, eventType)) {
                            boolean isContinue = canalProcess.processData(beforeCanalDataList, afterCanalDataList,eventType);
                            if (!isContinue){
                                break;
                            }

                        }
                    }
                }catch (Exception e){
                    log.error("数据解析失败={}",e.toString());
                }
            }
        }
    }

    private List<CanalData> buildColumn(String tableName, List<CanalEntry.Column> columns) {
        List<CanalData> canalDataList = Lists.newLinkedList();
        for (CanalEntry.Column column : columns) {
            CanalData canalData = new CanalData();
            canalData.setColumnName(column.getName());
            canalData.setValue(column.getValue());
            canalData.setUpdated(column.getUpdated());
            canalData.setTableName(tableName);
            canalDataList.add(canalData);
        }
        return canalDataList;
    }


    /**
     * 打印sql
     * @param entries
     * @throws Exception
     */
    private  void printSql(List<CanalEntry.Entry> entries) throws Exception {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }

            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            String tableName = entry.getHeader().getTableName();
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                String sql = getSql(rowChange.getEventType(),tableName,rowData);
               log.info("执行的sql={}",sql);
            }
        }
    }

    /**
     * 获取增删改的sql
     * @param eventType
     * @param tableName
     * @param rowData
     * @return
     */
    private String getSql(CanalEntry.EventType eventType, String tableName, CanalEntry.RowData rowData){
        String sql = null;
        switch (eventType) {
            case INSERT:
                sql = getInsertSql(tableName,rowData.getAfterColumnsList());
                break;
            case UPDATE:
                sql = getUpdateSql(tableName,rowData.getAfterColumnsList());
                break;
            case DELETE:
                sql = getDeleteSql(tableName,rowData.getBeforeColumnsList());
                break;
            default:
                break;
        }
        return sql;
    }

    private String getInsertSql(String tableName,List<CanalEntry.Column> columns){
        if(columns.size() == 0 || StringUtils.isBlank(tableName)){
            return null;
        }
        String keys = "";
        String values = "";
        for(int i=0;i<columns.size();i++){
            if(i != 0) {
                keys += ",";
                values += ",";
            }
            keys += columns.get(i).getName();
            values += getValue(columns.get(i));
        }
        String format = "INSERT INTO %s (%s) VALUES (%s)";
        return String.format(format,tableName,keys,values);
    }

    private String getUpdateSql(String tableName,List<CanalEntry.Column> columns){
        if(columns.size() == 0 || StringUtils.isBlank(tableName)){
            return null;
        }
        String sets = "";
        String where = "";
        for(CanalEntry.Column column : columns){
            if(column.getIsKey()){
                where = column.getName() + "=" + getValue(column);
                continue;
            }
            if(!StringUtils.isBlank(sets)) {
                sets += ",";
            }
            sets += column.getName() + "=" + getValue(column);
        }
        String format = "UPDATE %s SET %s WHERE %s";
        return String.format(format,tableName,sets,where);
    }

    private String getDeleteSql(String tableName,List<CanalEntry.Column> columns){
        if(columns.size() == 0 || StringUtils.isBlank(tableName)){
            return null;
        }
        String where = "";
        for(CanalEntry.Column column : columns){
            if(column.getIsKey()){
                where = column.getName() + "=" + getValue(column);
                continue;
            }
        }
        String format = "DELETE FROM %s WHERE %s";
        return String.format(format,tableName,where);
    }

    private String getValue(CanalEntry.Column column){
        if(column.getIsNull()){
            return "null";
        }
        return String.format("'%s'",column.getValue());
    }

}
